#include "net/base/connection.h"
#include <set>
#include <arpa/inet.h>
#include <errno.h>
#include <stdio.h>
#include <poll.h>
#include <string.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <unistd.h>
#include "base/callback.h"
#include "base/logging.h"
#include "base/scoped_ptr.h"
#include "thread/executor.h"
  
using std::set;

ServerConnection::ServerConnection(int port) : sfd_(-1), port_(port), backlog_(10) {
}

void ServerConnection::set_backlog(int backlog) {
  backlog_ = backlog;
}

void ServerConnection::SetPollRunner(PollRunner* poll_runner) {
  poll_runner_ = poll_runner;
  poll_runner->SetServerConnection(this);
}

bool ServerConnection::StartListen() {
  sockaddr_in a;

  if ((sfd_ = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
    // TODO
    perror("Could not create socket");
    return false;
  }
  /*
  if (setsockopt(s, SOL_SOCKET, SO_REUSEADDR,
        (char *) &yes, sizeof(yes)) == -1) {
    LOG(FATAL) << "Could create socket";
    perror("setsockopt");
    close(s);
    return -1;
  }*/
  memset(&a, 0, sizeof(a));
  a.sin_port = htons(port_);
  a.sin_family = AF_INET;
  if (bind(sfd_, (sockaddr *) &a, sizeof(a)) == -1) {
    perror("Could not bind");
    close(sfd_);
    sfd_ = -1;
    return false;
  }
  LOG(INFO) << "accepting connections on port " << port_;
  if (listen(sfd_, backlog_) == -1) {
    perror("Could not listen");
    close(sfd_);
    sfd_ = -1;
    return false;
  }
  return true;
}

void ServerConnection::Disconnect(int fd) {
  WriteLock lock(&clients_mutex_);
  if (clients_.find(fd) == clients_.end()) {
    return;
  }
  clients_[fd].status = DISCONNECTED;
  HandleDisconnect(fd);
}

void ServerConnection::CleanDisconnected() {
  WriteLock lock(&clients_mutex_);
  for (hash_map<int, ConnClient>::iterator iter = clients_.begin();
      iter != clients_.end(); ) {
    if (iter->second.status == DISCONNECTED) {
      close(iter->second.sockfd);
      clients_.erase(iter++);
    } else {
      ++iter;
    }
  }
}

PollRunner::PollRunner() {
}

void PollRunner::SetServerConnection(ServerConnection* conn) {
  conn_ = conn;
}

void PollRunner::Loop() {
  while (!exit_) {
    conn_->CleanDisconnected();
    scoped_array<pollfd> polls;
    int polls_size = 0;
    {
      ReadLock(&conn_->clients_mutex_);
      polls_size = conn_->clients_.size() + 1;
      polls.Reset(new pollfd[polls_size]);
      polls[0].fd = conn_->sfd_;
      polls[0].events = POLLIN | POLLPRI;
      int index = 1;
      for (hash_map<int, ConnClient>::iterator iter = conn_->clients_.begin();
          iter != conn_->clients_.end(); ++iter) {
        polls[index].fd = iter->second.sockfd;
        polls[index].events = POLLIN | POLLPRI | POLLRDHUP;
        {
          ReadLock lock(&iter->second.byte_stream_mutex);
          if (iter->second.byte_stream.size() > 0) {
            polls[index].events |= POLLOUT;
          }
        }
        ++index;
      }
    }
    int r = poll(polls.get(), polls_size, 2000);
    if (r == 0) {
      // Timeout.
      continue;
    }
    if (r == -1) {
      LOG(ERROR) << "Error poll";
      // TODO: errorno.
      continue;
    }
    for (int i = 0; i < polls_size; ++i) {
      if (i == 0) {
        if (polls[i].revents & POLLIN) {
          // New clients.
          sockaddr_in newclient;
          socklen_t size = sizeof(newclient);
          int newfd = accept(conn_->sfd_, (sockaddr*) &newclient, &size);
          if (newfd < 0) {
            // TODO:errorno
            LOG(ERROR) << "Error accepting client";
          } else {
            VLOG(1) << "Connected";
            conn_->HandleConnect(newfd);
            ConnClient newclient;
            newclient.sockfd = newfd;
            newclient.status = CONNECTED;
            {
              WriteLock lock(&conn_->clients_mutex_);
              CHECK(conn_->clients_.find(newfd) == conn_->clients_.end());
              conn_->clients_[newfd] = newclient;
            }
          }
        }
      } else {
        if (polls[i].revents & POLLIN) {
          // TODO: MSGOOB. Handle error.
          int size = 1024 * 1024;
          char* buffer = new char[size];
          size = recv(polls[i].fd, buffer, size, MSG_DONTWAIT);
          if (size < 0) {
            // TODO:errorno
            LOG(ERROR) << "Error receiving data";
            perror("erro");
            delete[] buffer;
          } else if (size == 0) {
            conn_->Disconnect(polls[i].fd);
            continue;
          } else {
            VLOG(1) << "message received";
            conn_->HandleRead(polls[i].fd, buffer, size);
          }
        }
        if (polls[i].revents & POLLRDHUP ||
            polls[i].revents & POLLHUP) {
          VLOG(1) << "Disconnect";
          conn_->Disconnect(polls[i].fd);
          continue;
        }
        if (polls[i].revents & POLLNVAL) {
          LOG(ERROR) << "Not open";
        }
        if (polls[i].revents & POLLOUT) {
          conn_->HandleWrite(polls[i].fd);
        }
      }
    }
  }
}

ConnClient* ServerConnection::GetClient(int fd) {
  ReadLock rl(&clients_mutex_);
  if (clients_.find(fd) == clients_.end()) {
    return NULL;
  }
  return &clients_[fd];
}

void ServerConnection::HandleWrite(int fd) {
  ConnClient* client = GetClient(fd);
  if (client == NULL) return;
  WriteLock lock(&client->byte_stream_mutex);
  if (client->byte_stream.size() == 0) return;
  int buff_size = std::min(2 << 16, client->byte_stream.size());
  char* buffer = new char[buff_size];
  client->byte_stream.Peek(buffer, buff_size);
  ssize_t result = send(fd, buffer, buff_size, MSG_NOSIGNAL);
  if (result == -1) {
    perror("Error sending");
  } else {
    client->byte_stream.Pop(result);
  }
}

bool ServerConnection::Send(int sockfd, const char* buffer, int size) {
  ConnClient* client = NULL;
  ReadLock rl(&clients_mutex_);
  if (clients_.find(sockfd) == clients_.end()) {
    return false;
  }
  client = &clients_[sockfd];
  if (client->status != CONNECTED) {
    return false;
  }
  WriteLock lock(&client->byte_stream_mutex);
  client->byte_stream.Push(buffer, size);
  return true;
}

void PollRunner::SetExit() {
  // TODO: mutex.
  exit_ = true;
}

